#include <moying_comm_lib/subscribe.h>
#include <thread>

namespace moying
{
namespace communication
{
MoyingCommuSubscribe::MoyingCommuSubscribe(const std::string &url,
                                     const std::string &identity) : socket_(MoyingCommuBase::ctx_, zmq::socket_type::sub)
{
  MoyingCommuBase::url_ = url;
  MoyingCommuBase::identity_ = identity;
}

MoyingCommuSubscribe::MoyingCommuSubscribe(const std::string &ip, const unsigned int port, const std::string &identity)
:MoyingCommuSubscribe(std::string("tcp://" + ip + ":" + std::to_string(port)), identity)
{
}

MoyingCommuSubscribe::MoyingCommuSubscribe(const std::string &content, const std::string &ip, const unsigned int port, const std::string &identity, RecvCall func, bool is_thread_safe)
:MoyingCommuSubscribe(ip, port, identity)
{
  setRecvCall(func, is_thread_safe);
}

MoyingCommuSubscribe::~MoyingCommuSubscribe()
{
  close();
}

bool MoyingCommuSubscribe::run()
{
  if (is_started_)
    return false;
  try
  {
    if(heartbeat_timeout_ms_ > 0){
      socket_.setsockopt(ZMQ_HEARTBEAT_IVL, heartbeat_interval_ms_);
      socket_.setsockopt(ZMQ_HEARTBEAT_TIMEOUT, heartbeat_timeout_ms_);
    }
    socket_.setsockopt(ZMQ_RECONNECT_IVL, kReconnIVL);
    socket_.setsockopt(ZMQ_RECONNECT_IVL_MAX, kReconnMax);
    socket_.setsockopt(ZMQ_LINGER, kLingerTimeout);

    std::ostringstream o;
    o<<serial_++;
    std::string inproc_name("inproc://monitor-subscribe");
    inproc_name += o.str();

    monitor_start_waiter_->set_signaled(false);
    createMonitor(socket_, inproc_name);
    if(!monitor_start_waiter_->signaled())  //防止notify后再wait
        monitor_start_waiter_->wait();

    socket_.connect(url_);
    createLoop();
  }
  catch (zmq::error_t e)
  {
    callLog(MoyingCommuLogLevels::FATAL, "MoyingCommuSubscribe::run() throw %s err_code:%d\n",
            e.what(), e.num());
    return false;
  }
  is_started_ = true;
  return true;
}

bool MoyingCommuSubscribe::close()
{

  if (is_stoped_)
      return false;

  callLog(MoyingCommuLogLevels::TRACE, "%s","enter subscribe mode socket close\n");
  is_stoped_ = true;
  while (is_started_)
  {
    SLEEP(10);
    if (is_loop_exit_ && is_monitor_exit_)
      break;
  }
  socket_.close();
  is_started_ = false;
  callLog(MoyingCommuLogLevels::TRACE, "%s","leave subscribe mode socket close\n");
  return true;
}

bool MoyingCommuSubscribe::unsubscribe(const std::string &content)
{
  if(!alterSubContent(content, false))
    return false;
  if(sub_cbs_.find(content) != sub_cbs_.end())
    sub_cbs_.erase(content);
  return true;
}

bool MoyingCommuSubscribe::alterSubContent(const std::string &content, bool add)
{
  if(add && sub_cbs_.find(content) != sub_cbs_.end())
  {
    printf("Invalid subscription. Confilicting subscription on topic %s", content.c_str());
    return false;
  }
  else if(!add && sub_cbs_.find(content) == sub_cbs_.end())
  {
    printf("Invalid unsubscription. Unregistered content topic %s", content.c_str());
    return false;
  }
  if (content.size() == 0)
  {
    printf("Empty content");
    return false;
  }

  std::string subscriber = packSubscriber(content);
  try
  {
      socket_.setsockopt(add ? ZMQ_SUBSCRIBE : ZMQ_UNSUBSCRIBE, subscriber.c_str(), subscriber.size());
  }
  catch (zmq::error_t e)
  {
    callLog(MoyingCommuLogLevels::FATAL, "MoyingCommuSubscribe::alterSubContent() throw %s err_code:%d\n",
            e.what(), e.num());
    return false;
  }
  return true;
}

/**
 * @brief printPackWrapper      打印通讯数据包
 * @param pack                  数据包
 * @return
 */
void MoyingCommuSubscribe::printPackWrapper(const std::string &content, bytes_ptr pack, int thread_id)
{
  if (print_pack_call)
  {
    if(proto_types_.find(content) == proto_types_.end())
    {
      // WARNING("Content "<<content<<" not found in registered proto types");
      return;
    }
    
    char* p = pack->data();
    int total_size  = GET_INT(p);
    int header_size = GET_INT(p);
    p = pack->data();
    bytes_ptr data = std::make_shared<bytes_vec>(p+header_size , p+total_size);
    OFFSET(p,8);
    int identity_len = GET_INT(p);
    std::string identity(p,identity_len);
    OFFSET(p,identity_len);
    int pack_id = GET_INT(p);
    long long timestamp = GET_LONGLONG(p);
    timestamp = ntohll(timestamp);
    
    std::string data_str;
    proto_types_[content]->ParseFromArray(data->data(),data->size());
    google::protobuf::TextFormat::PrintToString(*proto_types_[content], &data_str);

    // std::string fancy_msg = "\n\033[1m\033[30m[Moying]:\033[0m" + std::to_string(timestamp) 
    //             + " [Subscriber: " + identity + "-\033[35m" + content + "\033[0m-" + std::to_string(pack_id) + "]\n";
    std::string time_str = getDateTime(timestamp);
    std::string fancy_msg = "\n[Moying]:" + time_str 
                + " [Subscriber: " + identity + "-" + content + "-" + std::to_string(pack_id) + "]\n";
    print_pack_call(fancy_msg + data_str);
  }
}

/**
 * @brief createLoop            创建主功能循环线程
 * @return
 */
void MoyingCommuSubscribe::createLoop()
{
  auto pkg = [&]() -> void {
    auto thread_id = getTid();
    callLog(MoyingCommuLogLevels::INFO, "[tid:%d] started subscribe loop.\n", thread_id);

    std::vector<zmq_pollitem_t> poll_vec;
    poll_vec.push_back({socket_, 0, ZMQ_POLLIN, 0});
    is_loop_exit_ = false;

    while (!is_stoped_)
    {
      try
      {
        // 拉取订阅数据
        zmq::poll(poll_vec, poll_timeout_ms_);
        if (poll_vec[0].revents & ZMQ_POLLIN)
        {
          zmq::message_t msg_content;
          socket_.recv(&msg_content);
          std::string content(static_cast<char *>(msg_content.data()),
                              0, msg_content.size());
          content = unpackSubscriber(content);
          zmq::message_t msg_data;
          socket_.recv(&msg_data);

          // 解析收取数据包, 根据需要打印内容
          bytes_ptr pack_recv = std::make_shared<bytes_vec>((char*)msg_data.data(),(char*)msg_data.data()+msg_data.size());
          bytes_ptr decode_pack_recv = decodeRecvBuf(pack_recv);
          // 调用接收回调函数
          if(sub_cbs_.find(content) != sub_cbs_.end())
          {
            sub_cbs_[content].func_->processData(decode_pack_recv);
            if(sub_cbs_[content].log_)
              printPackWrapper(content, pack_recv, thread_id);
          }
        }
      }
      catch (zmq::error_t e)
      {
        callLog(MoyingCommuLogLevels::FATAL, "[tid:%d] MoyingCommuSubscribe loop throw %s err_code:%d\n",
                thread_id, e.what(), e.num());
        if (e.num() == ETERM) // 上下文已终止
          break;
      }
    } // while
    callLog(MoyingCommuLogLevels::INFO, "[tid:%d] stoped subscribe loop.\n", thread_id);
    is_loop_exit_ = true;
  };

  std::thread thr(pkg);
  thr.detach();
}

bool MoyingCommuSubscribe::subscribe(const std::string& content, CallbackBasePtr cb, bool log)
{
  if(!alterSubContent(content, true))
      return false;
  sub_cbs_[content] = SubData(content, cb, log);
}

} // namespace communication
} // namespace moying
